package confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.GetResponse;
import com.rabbitmq.client.MessageProperties;
import config.RabbitConfig;

import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * 批量确认
 *
 * @author tuyou
 * @date 2022/6/14 22:27
 */
public class BatchConfirm {

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {

        try (Connection connection = RabbitConfig.getConnection();
             Channel channel = connection.createChannel()) {

            String queue = channel.queueDeclare().getQueue();

            channel.confirmSelect();
            int total = 100;
            int msgCount = 0;
            //本地缓存队列，用于失败重试
            LinkedBlockingQueue<String> list = new LinkedBlockingQueue<>();
            while (true) {
                String msg = String.valueOf(msgCount);
                channel.basicPublish("", queue, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
                //将发送出去的消息存入缓存中，缓存可以是一个 ArrayList 或者 BlockingQueue 之类的
                list.add(msg);
                if (++msgCount >= total) {
                    if (!channel.waitForConfirms()) {
                        //将缓存中的消息重新发送
                        System.out.println("将缓存中的消息重新发送");
                    }
                    list.clear();
                    //将缓存中的消息清空
                    break;
                }
            }

            for (int i = 0; i < total; i++) {
                GetResponse resp = channel.basicGet(queue, true);
                System.out.println(new String(resp.getBody()));
            }
        }
    }
}